package plgutils

import (
	"dgo/MyProj/PrjHttpUtils"
	"dgo/goutils/utils"
	"dgo/plg/plgbase"
	"fmt"
	"gitee.com/ymofen/supervalue"
	"sync/atomic"
	"time"
)

/*
将数据打包成json数组进行传递
*/
type JsonPackPlg struct {
	pushFailNum int32
	PlgBase
	msglst       *utils.SyncCycleList
	cachelst     *utils.SyncQueue
	cacheMaxSize int

	packMaxNum int
	lastOutput string
	outputmsg  string
	interval   time.Duration
	monitorPub *utils.MonitorSpeedSizeRec
}

/*
只接受，json对象的字符串，或者json对象
*/
func (this *JsonPackPlg) PushData(data interface{}) {
	if this.ReleaseFlag == 1 {
		return
	}
	if str, ok := data.(string); ok {
		if int(this.cachelst.Count()) > this.cacheMaxSize {
			atomic.AddInt32(&this.pushFailNum, 1)
			return
		}
		this.cachelst.Push(str)
	} else if json, ok := data.(*wrapper.SuperValue); ok {
		if int(this.cachelst.Count()) > this.cacheMaxSize {
			atomic.AddInt32(&this.pushFailNum, 1)
			return
		}
		if json != nil {
			str := json.AsJSONStringEx(false, false)
			this.cachelst.Push(str)
		}
	}
}

func (this *JsonPackPlg) Pub(topic string, max int, args ...interface{}) int {
	if len(args) > 0 {
		this.PushData(args[0])
	}
	return 0
}

func (this *JsonPackPlg) DispatchData(channelid string, sender interface{}, data interface{}, tag interface{}) (err error) {
	this.PushData(data)
	return nil
}

func (this *JsonPackPlg) OutputData(str string, n int) {
	this.outputmsg = fmt.Sprintf("[%s]进行推送", utils.NowString())
	var packB utils.BytesBuilder
	packB.Appendf(`{"t":%d, "n":%d, "topic":"%s","list":[`, time.Now().Unix(), n, utils.EscapeJsonStrSimple(this.OutputEx.GetOutputTopic()))
	packB.WriteString(str)
	packB.WriteString("]}")
	packstr := packB.String()
	this.lastOutput = packstr
	this.OutputEx.DispatchData("", this, packstr, nil)
}

/*
		{
		  "id": "status-ex",
		  "type": "utils.json.pack",
		  "conf": {
			  "interval-secs":10,
	          "pack-max-num":10,
			  "output":{}
		  }
		}
*/
func (this *JsonPackPlg) ReloadConf(conf *wrapper.SuperValue) error {
	if conf == nil {
		return nil
	}

	if this.msglst == nil {
		this.msglst = utils.NewSyncCycleList()
	}

	if this.cachelst == nil {
		this.cachelst = utils.NewSyncQueue()
	}

	this.cacheMaxSize = conf.IntByName("maxcache", 10240)

	this.packMaxNum = conf.IntByName("pack-max-num", 10)
	if this.packMaxNum <= 0 {
		this.packMaxNum = 10
	}

	this.ReloadOutputConf(conf.ValueByName("output"))

	interval := conf.DurationByName("interval-secs", 5) * time.Second
	if interval < 0 {
		interval = time.Second * 5
	}
	this.interval = interval

	utils.DefaultPatrolTask().AddTask(this.PlugId, interval, func(id interface{}, args ...interface{}) {
		utils.DefaultWorkers().PostTaskFunc(this.PlugId, func() {
			t0 := time.Now()
			totalcnt := 0
			this.outputmsg = fmt.Sprintf("[%s]进行处理", utils.NowString())
			defer func() {
				this.outputmsg = fmt.Sprintf("[%s]处理完成, 耗时:%d(ms), 处理数量：%d", utils.NowString(), time.Since(t0).Milliseconds(), totalcnt)
			}()

			var lstB utils.BytesBuilder
			var popOnce = func() (remaincnt, cnt int, str string) {
				for {
					if ok, data := this.cachelst.Pop(); ok {
						if cnt > 0 {
							lstB.WriteString(",\n")
						}
						lstB.WriteString(data.(string))
						cnt++
						if cnt >= this.packMaxNum {
							break
						}
					} else {
						break
					}
				}
				remaincnt = int(this.cachelst.Count())
				str = lstB.String()
				return
			}

			for {
				r0, cnt, str := popOnce()
				if cnt > 0 {
					totalcnt += cnt
					this.OutputData(str, cnt)
					lstB.Cleanup()
				}
				if r0 == 0 {
					break
				}
			}

		})
	})
	return nil
}

func (this *JsonPackPlg) Close() (err error) {
	this.PlgBase.Close()
	utils.DefaultPatrolTask().DelTask(this.PlugId)
	return
}

func (this *JsonPackPlg) DoStatusRequestRequest(args ...interface{}) (resp interface{}) {
	json, _ := plgbase.NewSVArgs(false, args...)

	if json != nil {
		itype := json.IntByName("type", 0)
		if itype == 99 {
			json := wrapper.NewSVObject()
			json.SetKeyValue("cacheSize", this.cachelst.Count())
			json.SetKeyValue("cacheMaxSize", this.cacheMaxSize)
			json.SetKeyValue("cachePushFail", this.pushFailNum)
			json.SetKeyValue("interval", utils.HumanDuration(this.interval))
			json.SetKeyValue("outputmsg", this.outputmsg)
			json.SetKeyValue("output", this.OutputEx.StatusString())
			json.SetKeyValues("now", utils.NowString())
			return json
		} else if itype == 98 {
			return this.msglst
		} else if itype == 1 {
			return this.lastOutput
		} else {
			//resp = this.DoRequest(args...)
			//if resp != nil {
			//	return resp
			//}
		}
	}
	var sb utils.BytesBuilder
	PrjHttpUtils.AppendIndexBody(&sb, "status", "status?type=99")
	PrjHttpUtils.AppendIndexBody(&sb, "msglist", "status?type=98&n=100&searchval=")
	PrjHttpUtils.AppendIndexBody(&sb, "lastoutput", "status?type=1")
	return sb.Bytes()

}

func (this *JsonPackPlg) LogInfoMessage(s string, args ...interface{}) {
	this.msglst.LogMsg(s, args...)
	utils.Log.Msgf(this.PlugId, utils.LogLV_INFO, 1, s, args...)
}

func (this *JsonPackPlg) LogWarnMessage(s string, args ...interface{}) {
	this.msglst.LogMsg(s, args...)
	utils.Log.Msgf(this.PlugId, utils.LogLV_WARN, 1, s, args...)
}

func CreateJsonPackPlgFunc(args ...interface{}) (rval interface{}, err error) {
	obj := &JsonPackPlg{}

	if len(args) > 0 {
		if str, ok := args[0].(string); ok {
			obj.PlugId = str
		}
	}
	rval = obj
	return
}
